//package org.zfes.snowier.xyz.analysis.queue;
//
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//import java.util.concurrent.LinkedBlockingQueue;
//import java.util.concurrent.ThreadFactory;
//import java.util.concurrent.ThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.atomic.AtomicBoolean;
//
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.core.Cursor;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.zfes.snowier.common.eventpub.BasicMsg;
//import org.zfes.snowier.common.eventpub.EventBusBucket;
//import org.zfes.snowier.common.redis.RedisHelper;
//import com.google.common.util.concurrent.ThreadFactoryBuilder;
//
//public class RedisQueueBucket2  implements QueueBucket{
//
//	private static final String key_prefix="mo_li_";
//	@Autowired
//	private RedisTemplate<String,Object> redisTemplate;
//	
////	offer 
////	clear
//	
//	private static ExecutorService postExecutorService;
//	
//	private static ExecutorService offerExecutorService;
//	
//	private static final int maxOfferThreads=6;
//	private static final int maxOfferThreadQueueSize=1024*1024;
//	
//	
//	private static AtomicBoolean sheduelPoolStarted = new AtomicBoolean(false);
//	
//	
//	@Override
//	public  void offer(String topic,Object msg) {
//		startPostSheduel();
//		getOfferPool().submit(()->redisTemplate.opsForList().leftPush(key_prefix+topic,msg));
//	}
//	
//	
//	@Override
//	public void clear(String topic) {
//		redisTemplate.delete(key_prefix+topic);
//	}
//	
//	
//	private  void startPostSheduel(){
//		if(sheduelPoolStarted.get()) {
//			return;
//		}
//		if(postExecutorService==null) {
//			postExecutorService=Executors.newSingleThreadExecutor();
//		}
//		postExecutorService.submit(()->{
//			while(true) {
//				try {
//					   Cursor<String> cursor = RedisHelper.scan(redisTemplate, "*"+key_prefix+"*", 500);
//					   while (cursor.hasNext()) {
//						   
//					    	String key=new String(cursor.next().getBytes());
//					    	String topic=key.replace(key_prefix, "");
//					    	Object data =redisTemplate.opsForList().rightPop(key, 0, TimeUnit.NANOSECONDS);
//					    	if(data!=null) {
//								EventBusBucket.post(key, new BasicMsg(data).setTopic(topic));
//							}
//					    }
//					   RedisHelper.close(cursor);
//					
//				}catch (Exception e) {
//					e.printStackTrace();
//				}
//			}
//		});
//		sheduelPoolStarted.set(true);
//	
//	}
//	
//	
//	private static ExecutorService  getOfferPool() {
//		if(offerExecutorService != null) {
//			return offerExecutorService;
//		}
//		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("-queueBucket-OfferPool-pool-%d").build();
//		offerExecutorService = new ThreadPoolExecutor(0, maxOfferThreads,0L, 
//						TimeUnit.MILLISECONDS,
//						new LinkedBlockingQueue<Runnable>(maxOfferThreadQueueSize), 
//						namedThreadFactory, 
//						new ThreadPoolExecutor.AbortPolicy());
//		return offerExecutorService;				
//		
//	}
//
//	
//	
//}
